package org.unigram.oluolu.recommend;

import java.io.IOException;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Reducer;
import org.apache.hadoop.mapreduce.Mapper.Context;

/**
 *
 */
public class SelectClustersReducer extends
    Reducer<LongWritable, Text, LongWritable, Text>  {

    /**
     * reduce method.
     * 
     * @param clusterId cluster id
     * @param values set of users in the cluster  
     * @param context write the keys and values pairs.
     * @throws IOException -
     * @throws InterruptedException -
     */
    public final void reduce(final LongWritable clusterId,
            final Iterable<Text> values, final Context context)
            throws IOException, InterruptedException {
        
        StringBuffer idsStr = new StringBuffer();
        long clusterSize = 0;
        for (Text id : values) {
            idsStr.append(id.toString());
            idsStr.append(" ");
            clusterSize += 1;
            if (clusterSize >= this.maximumClusterSize) {
                break;
            }
        }
        
        if (this.minimumClusterSize <= clusterSize) {
            context.write(clusterId, new Text(idsStr.toString()));            
        }        
    }
    
    /**
     * setup.
     * 
     * @param context contains needed configuration.
     */
    @Override
    public final void setup(final Context context) {
        Configuration jc = context.getConfiguration();
        if (context == null || jc == null) {
            jc = new Configuration();
        }
        this.maximumClusterSize 
            = jc.getLong(RecommendConstants.MAX_CLUSTER_SIZE,
                    RecommendConstants.DEFAULT_MAX_CLUSTER_SIZE);
        this.minimumClusterSize 
            = jc.getLong(RecommendConstants.MIN_CLUSTER_SIZE,
                RecommendConstants.DEFAULT_MIN_CLUSTER_SIZE);
    }

    /** maximum number of examples in a cluster. */
    private long maximumClusterSize;
    
    /** minimum number of examples in a cluster. */    
    private long minimumClusterSize;    
}
